package com.taobao.analyse.main;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;

import com.taobao.analyse.map.ItemPriceChangedMapper;
import com.taobao.analyse.reduce.ItemPriceChangedReducer;
import com.taobao.cmp.lib.sst.mapred.input.SSTFileInputFormat;

/**
 * 价格变更记录统计入口
 * @author tieyi.qlr
 */
public class ItemPriceChangedMain {
	private String	baseOutPath	= "/group/tbauction/users/qianyi.zby/tieyi.qlr/output/";

	public JobConf getJob(String[] args) throws Exception {
		JobConf conf = new JobConf(new Configuration(), ArchiveMain.class);
		// MISC
		conf.setJobName("DupPriceChange");
		conf.setMaxMapAttempts(2);
		conf.setNumReduceTasks(1);

		// MAPPER
		conf.setMapOutputKeyClass(LongWritable.class);
		conf.setMapOutputValueClass(Text.class);

		// REDUCER
		conf.setOutputKeyClass(LongWritable.class);
		conf.setOutputValueClass(Text.class);
		//设置reducer
		conf.setReducerClass(ItemPriceChangedReducer.class);

		List<String> paths = new ArrayList<String>();
		paths.add("/group/tbauction/lifecycle/archive/20120601-02");
		
		// 加入SSTFile路径列表
		for (String x : paths) {
			System.out.println("添加SSTFile-Input" + x);
			//OUTPUT
			MultipleInputs.addInputPath(conf, new Path(x), SSTFileInputFormat.class, ItemPriceChangedMapper.class);
		}

		// OUTPUT
		Path out = this.buildOutputPath();
		conf.setOutputFormat(TextOutputFormat.class);

		System.out.println("设置输出路径:" + out.toString());
		TextOutputFormat.setOutputPath(conf, out);

		return conf;
	}

	private Path buildOutputPath() {
		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");
		String today = dateFormat.format(new Date());

		return new Path(baseOutPath + "pt=" + today + "000000/");
	}

	public static void main(String[] args) throws Exception {
		try {
			JobClient.runJob(new ItemPriceChangedMain().getJob(args));
		} catch (Exception e) {
			throw e;
		}
	}
}
